﻿using EventBus.Common;
using Microsoft.Extensions.DependencyInjection;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace EventBus.RabbitMQ
{
    public class ProcessCommandHandler : IProcessCommandHandler
    {
        private ConcurrentQueue<CommandHandlerPara<ICommand>> _commandHandlerQueue;
        private IServiceProvider _serviceProvider;
        private int _isUsing; //1使用 0未使用
        public ProcessCommandHandler(IServiceCollection serviceCollection)
        {
            _serviceProvider = serviceCollection.BuildServiceProvider();
            _commandHandlerQueue = new ConcurrentQueue<CommandHandlerPara<ICommand>>();
        }

        public void EnqueueMessage(CommandHandlerPara<ICommand> handlerPara)
        {
            _commandHandlerQueue.Enqueue(handlerPara);
            if (TryUsing())
            {
                Task.Factory.StartNew(CommandHandlerProcess);
            }
        }

        public async Task CommandHandlerProcess()
        {
            do
            {
                while (_commandHandlerQueue.TryDequeue(out CommandHandlerPara<ICommand> handlerPara))
                {
                    var command = SwifterJsonSerializer.DeserializeObjectByByte<ICommand>(handlerPara.ByteArray);
                    var commandHandlerImpl = _serviceProvider.GetService<ICommandHandler<ICommand>>();
                    await commandHandlerImpl.HandleAsync(command);
                    handlerPara.Channel.BasicAck(deliveryTag: handlerPara.DeliveryTag, multiple: false);
                }
                Interlocked.Exchange(ref _isUsing, 0);
            } while (!_commandHandlerQueue.IsEmpty && Interlocked.CompareExchange(ref _isUsing, 1, 0) == 0);
        }

        public bool TryUsing()
        {
            return Interlocked.CompareExchange(ref _isUsing, 1, 0) == 0;
        }

    }
}
